package rdd.partition;

import org.apache.spark.Partitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.List;

public class Spark95_Partitioner {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf();
        conf.setMaster("local[*]");
        conf.setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);

        final List<Tuple2<String,Integer>> datas = new ArrayList<>();
        datas.add(new Tuple2<String,Integer>("a",1));
        datas.add(new Tuple2<String,Integer>("a",2));
        datas.add(new Tuple2<String,Integer>("a",3));
        datas.add(new Tuple2<String,Integer>("a",4));

        final JavaRDD<Tuple2<String,Integer>> rdd = jsc.parallelize(datas,2);
        JavaPairRDD<String, Integer> maptopairRDD = rdd.mapToPair(t -> t);
        //传入自定义分区类实现自定义分区，此时没有重写equals和hashcode方法时，task的数量是5
        JavaPairRDD<String, Integer> reduceRDD = maptopairRDD.reduceByKey(new Mypartitioner(3), Integer::sum);
        //因为reduceByKey进行了两次，第二次reduceByKey按照道理来说是没有用的，因为是两次相同的shuffle，但是task却为8
        //是因为的reduceByKey中有一个scala语言的比较，对比两次的new Mypartitioner是否相同。
        //因为scala中==对于基础类型来说是比较内容，对于非基本类型来说是对比内存地址，这里两个都是new，所以内存地址不一样，不能判定为同一个shuffle 操作
        //如果想比较内容，需要对Mypartitioner中的equals方法和hashcode进行重写
        JavaPairRDD<String, Integer> reduceRDD1 = reduceRDD.reduceByKey(new Mypartitioner(3), Integer::sum);
    }
}

class Mypartitioner extends Partitioner {

    private int numPartitions;

    public Mypartitioner( int num ) {
        this.numPartitions = num ;
    }

    @Override
    //确定分区数量
    public int numPartitions() {
        return this.numPartitions;
    }

    @Override
    //确定位于哪个分区
    public int getPartition(Object key) {
        if ("nba".equals(key)){
            return 0;
        }else if ("wnba".equals(key)){
            return 1;
        }else {
            return 2;
        }
    }

    //加入下面两个代码，就是只有5个Task了

    //重写hashcode和eqals方法用于比较传入的自定义分区中的属性是否一致
    public int hashCode(){
        return numPartitions;
    }

    @Override
    public boolean equals(Object o) {
        if ( o instanceof Mypartitioner){
            Mypartitioner other = (Mypartitioner) o;
            return this.numPartitions == other.numPartitions;
        }else{
            return false;
        }
    }
}